package com.etc.wordcount;


import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * @author 上等的猪头肉
 * @Title: WCSpout
 * @ProjectName storm
 * @Description: TODO
 * @date 2018/11/15 16:01
 */
public class WCSpout extends BaseRichSpout {

    Map map;
    TopologyContext topologyContext;
    SpoutOutputCollector outputCollector;

    String[] file = new String[]{
            "hello hello word",
            "hello storm storm",
            "spark storm spark"
    };

    Random r = new Random();
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.map = map;
        this.topologyContext = topologyContext;
        this.outputCollector = spoutOutputCollector;
    }

    @Override
    public void nextTuple() {

//        for (String f : file) {
//            List list = new Values(f);
//            outputCollector.emit(list);
//        }

        String line = file[r.nextInt(file.length)];
        List tuple = new Values(line);

        this.outputCollector.emit(tuple);

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("line"));
    }
}
